
#include "core/job_manager.h"
#include "core/logs.h"
#include "core/utils.h"

#include <string.h>

/************************************************************************\
 * Job                                                                  *
\************************************************************************/

typedef struct
{
	VEJobState  *state;
	VEJobData    data;
	VEJobProcess process;
	VEJobSplit   split;
} VEJob;

static void ve_job_process( VEJob *job, int thread_id )
{
	job->process( &job->data, thread_id );
	ve_atomic_dec( &job->state->ref_count );
}

static int ve_job_split( VEJob *old_job, VEJob *new_job, int mode )
{
	memcpy( new_job, old_job, sizeof(VEJob) );

	if( new_job->split && new_job->split( &old_job->data, &new_job->data, mode ) )
	{
		ve_atomic_inc( &new_job->state->ref_count );
		return 1;
	}
	return 0;
}

/************************************************************************\
 * Job stack                                                            *
\************************************************************************/

typedef struct
{
	VEAtomicInt guard;
	int         position;
	VEJob       jobs[MAX_JOBS_PER_THREAD];
} VEJobStack;

static void ve_job_stack_init( VEJobStack *stack )
{
	stack->guard = 0;
	stack->position = 0;
}

static int ve_job_stack_push_top( VEJobStack *stack, const VEJob *job )
{
	int success = 0;
	ve_atomic_lock( &stack->guard );

	if( stack->position == MAX_JOBS_PER_THREAD )
		goto finally;

	memcpy( stack->jobs + stack->position, job, sizeof(VEJob) );
	stack->position++;
	success = 1;

finally:
	ve_atomic_unlock( &stack->guard );
	return success;
}

static int ve_job_stack_push_bottom( VEJobStack *stack, const VEJob *job )
{
	int success = 0;
	ve_atomic_lock( &stack->guard );

	if( stack->position == MAX_JOBS_PER_THREAD )
		goto finally;

	memmove( stack->jobs + 1, stack->jobs, stack->position*sizeof(VEJob) );
	memcpy( stack->jobs, job, sizeof(VEJob) );
	stack->position++;
	success = 1;

finally:
	ve_atomic_unlock( &stack->guard );
	return success;
}

static int ve_job_stack_pop( VEJobStack *stack, VEJob *job )
{
	int success = 0;
	ve_atomic_lock( &stack->guard );

	if( stack->position == 0 )
		goto finally;

	if( !ve_job_split( stack->jobs + stack->position - 1, job, JOB_SPLIT_MODE_POP ) )
		stack->position--;
	success = 1;

finally:
	ve_atomic_unlock( &stack->guard );
	return success;
}

static int ve_job_stack_give( VEJobStack *stack, VEJobStack *locked_stack )
{
	int success = 0;
	int count;
	if( !ve_atomic_try_lock( &stack->guard ) ) return 0;

	if( stack->position == 0 )
		goto finally;

	if( stack->position == 1 )
	{
		if( ve_job_split( stack->jobs, locked_stack->jobs + locked_stack->position, JOB_SPLIT_MODE_DIVIDE ) )
		{
			locked_stack->position++;
			success = 1;
		}
		goto finally;
	}

	count = stack->position >> 1;
	memcpy( locked_stack->jobs + locked_stack->position, stack->jobs, count*sizeof(VEJob) );
	locked_stack->position += count;
	memmove( stack->jobs, stack->jobs + count, (stack->position - count)*sizeof(VEJob) );
	stack->position -= count;
	success = 1;

finally:
	ve_atomic_unlock( &stack->guard );
	return success;
}

/************************************************************************\
 * Worker thread                                                        *
\************************************************************************/

typedef struct
{
	int            id;
	VEJobStack     jobs;
	VEThreadEvent *event;
	VEThread      *thread;
} VEWorkerThread;

int            g_worker_thread_count;
VEWorkerThread g_worker_threads[MAX_WORKER_THREAD_COUNT];
VEWorkerThread g_background_thread;
volatile int   g_worker_threads_running;

static int ve_worker_thread_process( VEWorkerThread *thread, int thread_id )
{
	VEJob job;
	int i;
	int offset;

	/* Try to perform own job */
	if( ve_job_stack_pop( &thread->jobs, &job ) )
	{
		ve_job_process( &job, thread_id );
		return 1;
	}

	/* Don't steal to background thread */
	if( thread->id < 0 ) return 0;

	/* Steal jobs from others */
	ve_atomic_lock( &thread->jobs.guard );
	offset = ve_get_time();
	for( i = 0; i < g_worker_thread_count; ++i )
	{
		int id = (i + offset) % g_worker_thread_count;
		if( id == thread->id ) continue;

		if( ve_job_stack_give( &g_worker_threads[id].jobs, &thread->jobs ) )
		{
			ve_atomic_unlock( &thread->jobs.guard );
			return 1;
		}
	}
	ve_atomic_unlock( &thread->jobs.guard );

	return 0;
}

static void ve_worker_thread( void *data )
{
	VEWorkerThread *thread = (VEWorkerThread*)data;

	while( g_worker_threads_running )
	{
		ve_thread_event_wait( thread->event );

		while( g_worker_threads_running &&
			  ve_worker_thread_process( thread, thread->id ) );
	}
}

static void ve_worker_thread_init( VEWorkerThread *thread, int id )
{
	thread->id = id;

	ve_job_stack_init( &thread->jobs );
	thread->event = ve_thread_event_create();
	thread->thread = ve_thread_create( ve_worker_thread, thread );
}

static void ve_worker_thread_done( VEWorkerThread *thread )
{
	ve_thread_delete( thread->thread );
	ve_thread_event_delete( thread->event );
}

static void ve_worker_thread_start( VEWorkerThread *thread, VEJob *job, int thread_id )
{
	if( job->state->priority == JOB_PRIORITY_LOW )
	{
		while( !ve_job_stack_push_bottom( &thread->jobs, job ) )
			ve_worker_thread_process( thread, thread_id );
	}
	else
	{
		if( !ve_job_stack_push_top( &thread->jobs, job ) )
			ve_job_process( job, thread_id );
	}
}

/************************************************************************\
 * Job manager                                                          *
\************************************************************************/

void ve_job_manager_init()
{
	int i;

	g_worker_threads_running = 1;

	g_worker_thread_count = ve_cpu_count() - 1;
	if( g_worker_thread_count < 1 )
		g_worker_thread_count = 1;

	for( i = 0; i < g_worker_thread_count; ++i )
		ve_worker_thread_init( &g_worker_threads[i], i+1 );
	ve_worker_thread_init( &g_background_thread, -1 );
}

void ve_job_manager_done()
{
	int i;

	g_worker_threads_running = 0;

	for( i = 0; i < g_worker_thread_count; ++i )
	{
		ve_thread_event_signal( g_worker_threads[i].event );
		ve_worker_thread_done( &g_worker_threads[i] );
	}
	ve_thread_event_signal( g_background_thread.event );
	ve_worker_thread_done( &g_background_thread );
}

int ve_worker_thread_count()
{
	return g_worker_thread_count + 1;
}

void ve_job_start( VEJobState *state, const VEJobData *data, VEJobProcess process, VEJobSplit split, int priority, int thread_id  )
{
	VEJob job;

	state->ref_count = 1;
	state->priority = priority;

	job.state = state;
	job.data = *data;
	job.process = process;
	job.split = split;

	if( state->priority == JOB_PRIORITY_BACKGROUND )
	{
		ve_worker_thread_start( &g_background_thread, &job, thread_id );
		ve_thread_event_signal( g_background_thread.event );
	}
	else
	{
		int i;

		int id = thread_id - 1;
		if( id < 0 ) id = ve_get_time() % g_worker_thread_count;
		ve_worker_thread_start( &g_worker_threads[id], &job, thread_id );

		for( i = 0; i < g_worker_thread_count; ++i )
			ve_thread_event_signal( g_worker_threads[i].event );
	}
}

void ve_job_wait( VEJobState *state, int thread_id )
{
	while( !ve_job_is_ready( state ) )
	{
		int id = thread_id - 1;
		if( id < 0 ) id = ve_get_time() % g_worker_thread_count;
		if( !ve_worker_thread_process( &g_worker_threads[id], thread_id ) )
			ve_worker_thread_process( &g_background_thread, thread_id );
	}
}

int ve_job_is_ready( VEJobState *state )
{
	return !state->ref_count;
}
